package org.huangrui.spark.scala.core.req

import org.apache.spark.util.AccumulatorV2
import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.mutable


/**
 * @Author hr
 * @Create 2024-10-19 3:00
 */
object HotCategoryTop10Analysis_3 {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setMaster("local[*]").setAppName("spark")
    val sc = new SparkContext(conf)
    // TODO 将文件作为数据源，对接RDD进行操作
    val data = sc.textFile("data/user_visit_action.txt")

    // TODO 1. 将多余的数据进行删除（过滤）处理
    val filterRdd = data.filter((line: String) => {
      val strings = line.split("_")
      "null".matches(strings(5))
    })
    val acc = new HotCategoryAccumulator()
    // 注册累加器到Spark中
    sc.register(acc, "hotCategory")
    filterRdd.foreach {
      action: String => {
        val datas = action.split("_")
        if (datas(6) != "-1" && datas(6) != "null"){
          // 点击的场合
          acc.add((datas(6), "click"))
        } else if (datas(8) != "null") {
          // 下单的场合
          val ids = datas(8).split(",")
          ids.foreach(
            (id: String) => {
              acc.add((id, "order"))
            })
        } else if (datas(10) != "null") {
          // 支付的场合
          datas(10).split(",").foreach((id: String) => {
            acc.add((id, "pay"))
          })
        } else {
          Nil
        }
      }
    }

    val accVal = acc.value
    val categories = accVal.values
    val hotCategories = categories.toList.sortWith((left: HotCategory, right: HotCategory) => {
      if (left.clickCnt > right.clickCnt) {
        true
      } else if (left.clickCnt == right.clickCnt) {
        if (left.orderCnt > right.orderCnt) {
          true
        } else if (left.orderCnt == right.orderCnt) {
          left.payCnt > right.payCnt
        } else {
          false
        }
      } else {
        false
      }
    })
    // 将结果采集到控制台打印出来
    hotCategories.take(10).foreach(println)

    sc.stop()
  }

  case class HotCategory(cid: String, var clickCnt: Int, var orderCnt: Int, var payCnt: Int)

  /**
   * 自定义累加器
   * 1. 继承AccumulatorV2，定义泛型
   * IN : ( 品类ID, 行为类型 )
   * OUT : mutable.Map[String, HotCategory]
   * 2. 重写方法（6）
   */
  class HotCategoryAccumulator extends AccumulatorV2[(String, String), mutable.Map[String, HotCategory]] {
    private val hcMap = mutable.Map[String, HotCategory]()

    override def isZero: Boolean = {
      hcMap.isEmpty
    }

    override def copy(): AccumulatorV2[(String, String), mutable.Map[String, HotCategory]] = new HotCategoryAccumulator()

    override def reset(): Unit = hcMap.clear()

    override def add(v: (String, String)): Unit = {
      val cid = v._1
      val actionType = v._2
      val category = hcMap.getOrElse(cid, HotCategory(cid, 0, 0, 0))
      actionType match {
        case "click" => category.clickCnt += 1
        case "order" => category.orderCnt += 1
        case "pay" => category.payCnt += 1
      }
      hcMap.update(cid, category)
    }

    override def merge(other: AccumulatorV2[(String, String), mutable.Map[String, HotCategory]]): Unit = {
      val map1 = this.hcMap
      val map2 = other.value
      map2.foreach {
        case (cid, hc) => {
          val category = map1.getOrElse(cid, HotCategory(cid, 0, 0, 0))
          category.clickCnt += hc.clickCnt
          category.orderCnt += hc.orderCnt
          category.payCnt += hc.payCnt
          map1.update(cid, category)
        }
      }
    }

    override def value: mutable.Map[String, HotCategory] = hcMap
  }
}
